package server;

import java.io.File;
import java.io.IOException;
import java.net.Socket;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;

import Messages.Message;
import Messages.MessageType;
import Messages.ResponseLockMessage;
import Utility.Constants;
import Utility.CheckAcceptProposalValue;
import Utility.CheckPrepareProposalValue;
import Utility.ClientUtilities;
import Utility.FileUtilities;
import Utility.Host;
import Utility.PaxosMessageReader;
import Utility.PaxosMessageReaderImpl;
import Utility.PaxosMessageWriter;
import Utility.PaxosMessageWriterImpl;
import Utility.ProposalIDGenerator;

public class PaxosServerThread extends Thread {

	private Socket clientSocket;
	private PaxosMessageReader paxosMessageReader;
	private PaxosMessageWriter paxosMessageWriter;
	private FileUtilities fileUtilities;
	private int serverId;
	private ProposalIDGenerator proposerIdGenerator;
	private CheckPrepareProposalValue checkPrepareProposalValue;
	private CheckAcceptProposalValue checkAcceptProposalValue;
	private boolean continueProcessingAcceptorRequest;

	public PaxosServerThread(Socket clientSocket,int serverId,boolean continueProcessingRequest) {
		this.clientSocket = clientSocket;
		this.serverId = serverId;
		this.paxosMessageWriter = new PaxosMessageWriterImpl();
		this.paxosMessageReader = new PaxosMessageReaderImpl();
		this.proposerIdGenerator = new ProposalIDGenerator();
		this.checkPrepareProposalValue = new CheckPrepareProposalValue();
		this.checkAcceptProposalValue = new CheckAcceptProposalValue();
		this.continueProcessingAcceptorRequest = continueProcessingRequest;
		this.fileUtilities = new FileUtilities();
	}

	@Override
	public void run() {

		// READ THE MESSAGE FROM SOCKET
		Message readMessage = null;
		int iterator;
		System.out.println("Initializing input and output object streams... ");
		try {
			System.out.println(ClientUtilities.getMessageSeparator());
			readMessage = paxosMessageReader.readMessage(clientSocket);
			if (readMessage == null)
				return;
			System.out.println("Read message of Type :" + readMessage.getMessageType());
		} catch (Exception e) {
			e.printStackTrace();
		}
		System.out.println(ClientUtilities.getMessageSeparator());
		MessageType messageType = readMessage.getMessageType();
		//System.out.println("Applying logic to Switch ... with messageType : " + messageType.name());
		
		switch (messageType) {

		/*******************************************************
		 * CLIENT RELATED LOGIC
		 ******************************************************/
		case REQUEST_LOCK:

			int lockValue = readMessage.getLockValue();
			File f = new File(System.getProperty("user.dir")+File.separator+"locksfile"+serverId);
			if(!f.exists()){
				try {
					f.createNewFile();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}   
				HashMap<Integer, String> myMap = new HashMap<Integer, String>();
				myMap = fileUtilities.readFileAsMap(System.getProperty("user.dir")+File.separator+"locksfile"+serverId);
				if(myMap.containsKey(lockValue)){
					ResponseLockMessage responseMessage = new ResponseLockMessage();
					responseMessage.setMessageType(MessageType.RESPONSE_LOCK);
					responseMessage.setLockGranted(false);
					paxosMessageWriter.writeMessage(clientSocket, responseMessage);
				}  else{
					/**
					 * GET LIST OF ACCEPTOR CLIENT SOCKETS
					 */
					System.out.println(ClientUtilities.getMessageSeparator());
					List<Socket> acceptorClientSockets = getAcceptorClientSockets(serverId);
					int totalNumberOfAcceptors = acceptorClientSockets.size();
					System.out.println("Total number of acceptors  = " +totalNumberOfAcceptors);

					/**
					 * SEND PREPARE PROPOSAL Message
					 */
					System.out.println(ClientUtilities.getMessageSeparator());
					int proposalID = proposerIdGenerator.generateProposalID(System.getProperty("user.dir")+File.separator+serverId, Constants.initialProposalId, readMessage, serverId);
					Message prepareProposalMessage = ClientUtilities.getMessage(MessageType.PREPARE_PROPOSAL, proposalID, lockValue);
					writeMessages(acceptorClientSockets, prepareProposalMessage);
					List<Message> proposalAcksOrNacks =  new ArrayList<Message>();

					/**
					 * RECIEVE PROMISE ACK/NACK now..
					 */
					for(Socket s:acceptorClientSockets)
			        {
			        	try {
							s.setSoTimeout(100);
						} catch (Exception e) {
							// TODO Auto-generated catch block
							System.out.println("Acceptor Timedout");
							//e.printStackTrace();					
						}
			        }     
					System.out.println(ClientUtilities.getMessageSeparator());
					proposalAcksOrNacks = readMessages(acceptorClientSockets);
					int totalNumberOfPromiseAcks = 0;
					int totalNumberOfPromiseNacks = 0;
					int highestProposalId = 0;
					for(Message m : proposalAcksOrNacks ){
						if(m!=null){
						if(m.getMessageType()==MessageType.PROMISE_ACK){
							totalNumberOfPromiseAcks++;
						}else if(m.getMessageType()==MessageType.PROMISE_NACK){
							totalNumberOfPromiseNacks++;
						}
						if(m.getProposalNumber()>highestProposalId){
							highestProposalId = m.getProposalNumber();
						}
						}
					}

					List<Socket> listSockets = new ArrayList<Socket>();
					listSockets = getAcceptorClientSockets(serverId);
					/**
					 * SEND ACCEPT_REQUEST Message if NumbersofPromisesAcks > N/2
					 * SEND RE-PROPOSE if NumberofPromiseNacks < N/2
					 */
					System.out.println(ClientUtilities.getMessageSeparator());
					if(totalNumberOfPromiseAcks>(totalNumberOfAcceptors/2)){
						Message acceptRequestMessage = ClientUtilities.getMessage(MessageType.ACCEPT_REQUEST, proposalID, lockValue);
						writeMessages(listSockets, acceptRequestMessage);
					}else{
						//Regenerate the prepare message with highest proposal-id
						Message prepareHighestProposalMessage = ClientUtilities.getMessage(MessageType.PREPARE_PROPOSAL, highestProposalId+100, lockValue);
						writeMessages(listSockets, prepareHighestProposalMessage);
					}

					/**
					 * RECIVE ACCEPT ACK/NACK message
					 */

					System.out.println(ClientUtilities.getMessageSeparator());
					List<Message> acceptAcksOrNacks = new ArrayList<Message>();
					acceptAcksOrNacks = readMessages(listSockets);
					System.out.println("Total number of ACCEPT_REQUEST Acknowledgment/Nacks recived : " + acceptAcksOrNacks.size());
					int totalAcceptAcks = 0;
					int totalAcceptNacks = 0;
					for( Message m : acceptAcksOrNacks){
						System.out.println("RECIVED :" + m.getMessageType().name());
						if(m.getMessageType()==MessageType.ACCEPT_ACK){
							totalAcceptAcks++;
						}else{
							totalAcceptNacks++;
						}
					}
					List<Socket> listOfSocket = new ArrayList<Socket>();
					listOfSocket = getAcceptorClientSockets(serverId);

					/**
					 * SEND LEARN MESSAGE IF NUMBER OF ACCEPT ACKS ARE GREATER THAN N/2
					 */
					System.out.println(ClientUtilities.getMessageSeparator());
					if(totalAcceptAcks>= (totalNumberOfAcceptors/2)){
						Message learnMessage =	ClientUtilities.getMessage(MessageType.LEARN,acceptAcksOrNacks.get(0).getProposalNumber(), acceptAcksOrNacks.get(0).getLockValue());
						learnMessage.setClientId(readMessage.getClientId());
						writeMessages(listOfSocket, learnMessage);
						
						// Updating  Locks file
						System.out.println("Writing lock value and client to LockFile");
						String text = learnMessage.getLockValue()+","+readMessage.getClientId();
						FileUtilities.writeToFile(System.getProperty("user.dir")+File.separator+"locksfile"+serverId, text, true);
						
						// Updating proposal file
						System.out.println("Updating the proposal file with lockvalue : " + learnMessage.getLockValue() + "with proposal iD as : " + learnMessage.getProposalNumber());
						String proposalText = learnMessage.getProposalNumber() + "," + learnMessage.getLockValue();
						FileUtilities.writeToFile(System.getProperty("user.dir")+File.separator + this.serverId,proposalText, false);
						
						//Send Response to client if its granted successfully
						ResponseLockMessage responseLock =new ResponseLockMessage();
						responseLock.setLockGranted(true);
						responseLock.setLockValue(learnMessage.getLockValue());
						responseLock.setMessageType(MessageType.RESPONSE_LOCK);
						paxosMessageWriter.writeMessage(clientSocket,responseLock);
					}

				}
			// Add paxos logic
			break;

		case RELEASE_LOCK:

			break;

			/*******************************************************
			 * ACCEPTOR RELATED STUFF
			 ******************************************************/
		case PREPARE_PROPOSAL:
			if(continueProcessingAcceptorRequest)
			{
				System.out.println(ClientUtilities.getMessageSeparator());
				System.out.println("Received " + MessageType.PREPARE_PROPOSAL.name());			
				Message proposalReplyMessage = checkPrepareProposalValue.checkMyFileForProposalValue(System.getProperty("user.dir")+File.separator+serverId, readMessage);
				paxosMessageWriter.writeMessage(clientSocket, proposalReplyMessage);
			}
			else
			{
				System.out.println("Acceptor Failed to Run");	
			}
			break;

		case ACCEPT_REQUEST:
			// Add paxos logic
			System.out.println(ClientUtilities.getMessageSeparator());
			System.out.println("Recived message of type : " + MessageType.ACCEPT_REQUEST);
			System.out.println("Message Proposal ID received : " + readMessage.getProposalNumber());
			System.out.println("Message Lock Value : " + readMessage.getLockValue());
			Message acceptReplyMessage = checkAcceptProposalValue.checkMyFileForProposalValue(System.getProperty("user.dir")+File.separator+serverId, readMessage);
			paxosMessageWriter.writeMessage(clientSocket, acceptReplyMessage);
			System.out.println(ClientUtilities.getMessageSeparator());
			break;

		case RESPONSE_LOCK:

			// TODO : I dont think we need this here. Investigate more....

			break;

			/*******************************************************
			 * LEARNER RELATED LOGIC
			 ******************************************************/
		case LEARN:
			System.out.println(ClientUtilities.getMessageSeparator());
			int learnedProposalId = readMessage.getProposalNumber();
			int learnedLockValue = readMessage.getLockValue();
			System.out.println("ProposalID received from LEARN Message :" + learnedProposalId);
			System.out.println("Lock value  received from LEARN Message : " + learnedLockValue);
			String proposalIdAndLockValue = null;
			try {
				proposalIdAndLockValue = FileUtilities.readFileAsString(System.getProperty("user.dir")+File.separator+serverId);
			} catch (IOException e) {
				e.printStackTrace();
			}
			String[] actualProposalIdAndLockValue = proposalIdAndLockValue.split(",");
			System.out.println("ProposalID in LOCAL :" + actualProposalIdAndLockValue[0]);
			System.out.println("Lock value in LOCAL : " + actualProposalIdAndLockValue[1]);

			String text = learnedProposalId+","+learnedLockValue;
			if(Integer.parseInt(actualProposalIdAndLockValue[0])!=learnedProposalId){
				FileUtilities.writeToFile(System.getProperty("user.dir")+File.separator+serverId, text,false);
				System.out.println("The server has updated learnt message successfully");
			}else if(Integer.parseInt(actualProposalIdAndLockValue[0])==learnedProposalId){
				System.out.println("The server has learnt successfully");
			}

         //writing into lock files.
			System.out.println("Writing lock value and client to LockFile");
			String text1 = readMessage.getLockValue()+","+readMessage.getClientId();
			FileUtilities.writeToFile(System.getProperty("user.dir")+File.separator+"locksfile"+serverId, text1, true);
			
			
			break;

		default:
			System.err.println(" Unsupported Message received. Unable to process the message :  "
					+ messageType.name());
		}
	}
	public void writeMessages(List<Socket> sockets,Message message){
		for(Socket s : sockets){
			System.out.println("Writing Message to : " + s.getInetAddress() + " with port as : "+ s.getPort() );
			paxosMessageWriter.writeMessage(s, message);
		}
	}

	public List<Message> readMessages(List<Socket> sockets){
		List<Message> messagesList = new ArrayList<Message>();
		for(Socket s : sockets){
			try{
			System.out.println("Reading Message from : " + s.getInetAddress() + " with port as : "+ s.getPort() );
			messagesList.add(paxosMessageReader.readMessage(s));
			}
			catch(Exception e)
			{
				System.out.println("Cannot read the message");
			}
		}
		return messagesList;
	}

	public List<Socket> getAcceptorClientSockets(int serverId){
		/**
		 * TODO task list
		 */
		//Create clientSockets to all other Servers. And create ObjectInput and ObjectOutputStreams for all servers/acceptors
		LinkedList<Host> acceptorsServersList =  ClientUtilities.getAllHostExcludingMe(serverId);
		LinkedList<Socket> acceptorClientSockets = new LinkedList<Socket>();
		for(Host h:acceptorsServersList){
			try {
				System.out.println("Creating client socket for Acceptor with hostname : " + h.getHostName()+ "with port number " + h.getPortNumber() );
				Socket cs = new Socket(h.getHostName(),Integer.parseInt(h.getPortNumber()));
				acceptorClientSockets.add(cs);
			} catch (NumberFormatException e) {
				e.printStackTrace();
			} catch (UnknownHostException e) {
				e.printStackTrace();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		return acceptorClientSockets;
	}
}
